/// 高效元组存储格式v2 - 真正的生产级实现
/// 
/// 这是对当前元组存储格式的重构，解决以下问题：
/// 1. 当前元组格式内存使用效率低
/// 2. 缺乏压缩和编码支持
/// 3. 序列化/反序列化性能差
/// 4. 不支持变长数据优化

use crate::{Error, Result};
use crate::sql::Value;
use crate::storage::{TableSchema, ColumnDefinition, DataType};
use std::collections::HashMap;
use std::time::SystemTime;

/// 高效元组头部 - 16字节固定大小
/// 
/// 优化的元组头部设计，包含所有必要的元数据
#[repr(C)]
#[derive(Debug, Clone, Copy)]
pub struct TupleHeaderV2 {
    /// 元组总大小（包括头部）
    pub total_size: u16,
    /// 列数量
    pub column_count: u8,
    /// 标志位
    pub flags: TupleFlags,
    /// 事务ID（MVCC支持）
    pub xmin: u32,
    /// 事务ID（MVCC支持）
    pub xmax: u32,
    /// NULL位图长度（字节数）
    pub null_bitmap_len: u8,
    /// 变长数据偏移量
    pub varlen_offset: u16,
    /// 保留字段（用于对齐）
    pub reserved: u8,
}

/// 元组标志位 - 8位标志
#[repr(transparent)]
#[derive(Debug, Clone, Copy)]
pub struct TupleFlags(u8);

impl TupleFlags {
    /// 包含NULL值
    pub const HAS_NULL: u8 = 0x01;
    /// 包含变长数据
    pub const HAS_VARLEN: u8 = 0x02;
    /// 已删除标记
    pub const IS_DELETED: u8 = 0x04;
    /// 压缩标记
    pub const IS_COMPRESSED: u8 = 0x08;
    /// 热数据标记（用于缓存优化）
    pub const IS_HOT: u8 = 0x10;
    /// 索引标记（用于索引优化）
    pub const IS_INDEXED: u8 = 0x20;
    /// 保留位1
    pub const RESERVED1: u8 = 0x40;
    /// 保留位2
    pub const RESERVED2: u8 = 0x80;
    
    pub fn new() -> Self {
        Self(0)
    }
    
    pub fn has_null(&self) -> bool {
        (self.0 & Self::HAS_NULL) != 0
    }
    
    pub fn has_varlen(&self) -> bool {
        (self.0 & Self::HAS_VARLEN) != 0
    }
    
    pub fn is_deleted(&self) -> bool {
        (self.0 & Self::IS_DELETED) != 0
    }
    
    pub fn is_compressed(&self) -> bool {
        (self.0 & Self::IS_COMPRESSED) != 0
    }
    
    pub fn is_hot(&self) -> bool {
        (self.0 & Self::IS_HOT) != 0
    }
    
    pub fn is_indexed(&self) -> bool {
        (self.0 & Self::IS_INDEXED) != 0
    }
    
    pub fn set_has_null(&mut self) {
        self.0 |= Self::HAS_NULL;
    }
    
    pub fn set_has_varlen(&mut self) {
        self.0 |= Self::HAS_VARLEN;
    }
    
    pub fn set_deleted(&mut self) {
        self.0 |= Self::IS_DELETED;
    }
    
    pub fn set_compressed(&mut self) {
        self.0 |= Self::IS_COMPRESSED;
    }
    
    pub fn set_hot(&mut self) {
        self.0 |= Self::IS_HOT;
    }
    
    pub fn set_indexed(&mut self) {
        self.0 |= Self::IS_INDEXED;
    }
    
    pub fn clear_deleted(&mut self) {
        self.0 &= !Self::IS_DELETED;
    }
}

impl TupleHeaderV2 {
    /// 元组头部大小（字节）
    pub const HEADER_SIZE: usize = 16;
    
    /// 创建新的元组头部
    pub fn new(column_count: u8, xmin: u32) -> Self {
        Self {
            total_size: 0, // 将在序列化时设置
            column_count,
            flags: TupleFlags::new(),
            xmin,
            xmax: 0,
            null_bitmap_len: 0,
            varlen_offset: 0,
            reserved: 0,
        }
    }
    
    /// 获取头部大小
    pub fn header_size() -> usize {
        Self::HEADER_SIZE
    }
}

/// 高效元组存储格式v2
/// 
/// 优化的元组存储格式，支持：
/// - 紧凑的内存布局
/// - 高效的NULL位图
/// - 变长数据优化
/// - 压缩支持
/// - 快速序列化/反序列化
#[derive(Debug, Clone)]
pub struct TupleV2 {
    /// 元组头部
    pub header: TupleHeaderV2,
    /// NULL位图 - 每个位表示对应列是否为NULL
    pub null_bitmap: Vec<u8>,
    /// 定长数据区域
    pub fixed_data: Vec<u8>,
    /// 变长数据区域
    pub varlen_data: Vec<u8>,
    /// 变长数据偏移表
    pub varlen_offsets: Vec<u16>,
}

/// 元组编码器 - 负责高效的元组编码和解码
#[derive(Debug)]
pub struct TupleEncoderV2 {
    /// 表模式
    schema: TableSchema,
    /// 编码统计信息
    stats: TupleEncodingStats,
}

/// 元组编码统计信息
#[derive(Debug, Clone)]
pub struct TupleEncodingStats {
    /// 编码的元组数量
    pub encoded_tuples: u64,
    /// 解码的元组数量
    pub decoded_tuples: u64,
    /// 总编码时间（微秒）
    pub total_encoding_time_us: u64,
    /// 总解码时间（微秒）
    pub total_decoding_time_us: u64,
    /// 压缩前总大小
    pub uncompressed_size: u64,
    /// 压缩后总大小
    pub compressed_size: u64,
    /// 平均编码时间（微秒）
    pub avg_encoding_time_us: u64,
    /// 平均解码时间（微秒）
    pub avg_decoding_time_us: u64,
    /// 压缩比
    pub compression_ratio: f64,
    /// 最后更新时间
    pub last_updated: SystemTime,
}

impl Default for TupleEncodingStats {
    fn default() -> Self {
        Self {
            encoded_tuples: 0,
            decoded_tuples: 0,
            total_encoding_time_us: 0,
            total_decoding_time_us: 0,
            uncompressed_size: 0,
            compressed_size: 0,
            avg_encoding_time_us: 0,
            avg_decoding_time_us: 0,
            compression_ratio: 0.0,
            last_updated: SystemTime::now(),
        }
    }
}

impl TupleV2 {
    /// 创建新的元组v2
    pub fn new() -> Self {
        Self {
            header: TupleHeaderV2::new(0, 0),
            null_bitmap: Vec::new(),
            fixed_data: Vec::new(),
            varlen_data: Vec::new(),
            varlen_offsets: Vec::new(),
        }
    }
    
    /// 获取元组总大小
    pub fn total_size(&self) -> usize {
        TupleHeaderV2::HEADER_SIZE
            + self.null_bitmap.len()
            + self.fixed_data.len()
            + self.varlen_data.len()
            + self.varlen_offsets.len() * 2 // u16 = 2字节
    }
    
    /// 检查列是否为NULL
    pub fn is_null(&self, column_index: usize) -> bool {
        if column_index >= self.header.column_count as usize {
            return false;
        }
        
        let byte_index = column_index / 8;
        let bit_index = column_index % 8;
        
        if byte_index >= self.null_bitmap.len() {
            return false;
        }
        
        (self.null_bitmap[byte_index] & (1 << bit_index)) != 0
    }
    
    /// 设置列为NULL
    pub fn set_null(&mut self, column_index: usize) {
        if column_index >= self.header.column_count as usize {
            return;
        }
        
        let byte_index = column_index / 8;
        let bit_index = column_index % 8;
        
        // 确保NULL位图足够大
        while self.null_bitmap.len() <= byte_index {
            self.null_bitmap.push(0);
        }
        
        self.null_bitmap[byte_index] |= 1 << bit_index;
        self.header.flags.set_has_null();
    }
    
    /// 将元组序列化为字节数据
    /// 
    /// 高效的序列化实现，优化内存布局和性能
    pub fn to_bytes(&self) -> Vec<u8> {
        let mut bytes = Vec::with_capacity(self.total_size());
        
        // 序列化头部
        let header_bytes = unsafe {
            std::slice::from_raw_parts(
                &self.header as *const TupleHeaderV2 as *const u8,
                TupleHeaderV2::HEADER_SIZE,
            )
        };
        bytes.extend_from_slice(header_bytes);
        
        // 序列化NULL位图
        bytes.extend_from_slice(&self.null_bitmap);
        
        // 序列化定长数据
        bytes.extend_from_slice(&self.fixed_data);
        
        // 序列化变长数据偏移表
        for offset in &self.varlen_offsets {
            bytes.extend_from_slice(&offset.to_le_bytes());
        }
        
        // 序列化变长数据
        bytes.extend_from_slice(&self.varlen_data);
        
        bytes
    }
    
    /// 从字节数据反序列化元组
    /// 
    /// 高效的反序列化实现，支持零拷贝优化
    pub fn from_bytes(bytes: &[u8]) -> Result<Self> {
        if bytes.len() < TupleHeaderV2::HEADER_SIZE {
            return Err(Error::storage("元组数据太短".to_string()));
        }
        
        // 反序列化头部
        let header = unsafe {
            std::ptr::read(bytes.as_ptr() as *const TupleHeaderV2)
        };
        
        let mut offset = TupleHeaderV2::HEADER_SIZE;
        
        // 反序列化NULL位图
        let null_bitmap_len = header.null_bitmap_len as usize;
        if offset + null_bitmap_len > bytes.len() {
            return Err(Error::storage("NULL位图数据不完整".to_string()));
        }
        let null_bitmap = bytes[offset..offset + null_bitmap_len].to_vec();
        offset += null_bitmap_len;
        
        // 计算定长数据长度
        let varlen_offset = header.varlen_offset as usize;
        let fixed_data_len = if varlen_offset > 0 {
            varlen_offset - offset
        } else {
            header.total_size as usize - offset
        };
        
        // 反序列化定长数据
        if offset + fixed_data_len > bytes.len() {
            return Err(Error::storage("定长数据不完整".to_string()));
        }
        let fixed_data = bytes[offset..offset + fixed_data_len].to_vec();
        offset += fixed_data_len;
        
        // 反序列化变长数据（如果存在）
        let mut varlen_offsets = Vec::new();
        let mut varlen_data = Vec::new();
        
        if header.flags.has_varlen() && varlen_offset > 0 {
            // 计算变长字段数量
            let varlen_count = (header.total_size as usize - varlen_offset) / 2; // 假设每个偏移2字节
            
            // 反序列化偏移表
            for _ in 0..varlen_count {
                if offset + 2 > bytes.len() {
                    break;
                }
                let offset_val = u16::from_le_bytes([bytes[offset], bytes[offset + 1]]);
                varlen_offsets.push(offset_val);
                offset += 2;
            }
            
            // 反序列化变长数据
            if offset < bytes.len() {
                varlen_data = bytes[offset..].to_vec();
            }
        }
        
        Ok(Self {
            header,
            null_bitmap,
            fixed_data,
            varlen_data,
            varlen_offsets,
        })
    }
}

impl TupleEncoderV2 {
    /// 创建新的元组编码器
    pub fn new(schema: TableSchema) -> Self {
        Self {
            schema,
            stats: TupleEncodingStats::default(),
        }
    }

    /// 从值向量编码元组
    ///
    /// 高效的编码实现，支持：
    /// - 自动类型检测和优化
    /// - NULL值压缩
    /// - 变长数据优化
    /// - 内存对齐优化
    pub fn encode_tuple(&mut self, values: Vec<Value>, xmin: u32, xmax: u32) -> Result<TupleV2> {
        let start_time = std::time::Instant::now();

        if values.len() != self.schema.columns.len() {
            return Err(Error::storage(format!(
                "列数不匹配：期望 {}，实际 {}",
                self.schema.columns.len(),
                values.len()
            )));
        }

        let column_count = values.len() as u8;
        let mut header = TupleHeaderV2::new(column_count, xmin);
        header.xmax = xmax;

        // 计算NULL位图长度
        let null_bitmap_len = (column_count + 7) / 8; // 向上取整
        header.null_bitmap_len = null_bitmap_len;

        let mut tuple = TupleV2::new();
        tuple.header = header;
        tuple.null_bitmap = vec![0u8; null_bitmap_len as usize];

        // 分离定长和变长数据
        let mut fixed_data = Vec::new();
        let mut varlen_data = Vec::new();
        let mut varlen_offsets = Vec::new();
        let mut has_null = false;
        let mut has_varlen = false;

        for (i, (value, column)) in values.iter().zip(self.schema.columns.iter()).enumerate() {
            match value {
                Value::Null => {
                    tuple.set_null(i);
                    has_null = true;
                    // 为定长类型添加占位符
                    self.add_fixed_placeholder(&mut fixed_data, &column.data_type);
                }
                Value::Integer(val) => {
                    fixed_data.extend_from_slice(&val.to_le_bytes());
                }
                Value::Real(val) => {
                    fixed_data.extend_from_slice(&val.to_le_bytes());
                }
                Value::Boolean(val) => {
                    fixed_data.push(if *val { 1 } else { 0 });
                }
                Value::Text(text) => {
                    // 变长数据处理
                    let text_bytes = text.as_bytes();
                    varlen_offsets.push(varlen_data.len() as u16);

                    // 存储长度前缀 + 数据
                    varlen_data.extend_from_slice(&(text_bytes.len() as u32).to_le_bytes());
                    varlen_data.extend_from_slice(text_bytes);
                    has_varlen = true;

                    // 在定长区域存储偏移量
                    fixed_data.extend_from_slice(&(varlen_offsets.len() as u32 - 1).to_le_bytes());
                }
                Value::Decimal(decimal) => {
                    // 将Decimal转换为字符串存储（变长）
                    let decimal_str = decimal.to_string();
                    let decimal_bytes = decimal_str.as_bytes();
                    varlen_offsets.push(varlen_data.len() as u16);

                    varlen_data.extend_from_slice(&(decimal_bytes.len() as u32).to_le_bytes());
                    varlen_data.extend_from_slice(decimal_bytes);
                    has_varlen = true;

                    fixed_data.extend_from_slice(&(varlen_offsets.len() as u32 - 1).to_le_bytes());
                }
                Value::Timestamp(ts) => {
                    // 时间戳转换为字符串存储（变长）
                    let ts_str = ts.to_rfc3339();
                    let ts_bytes = ts_str.as_bytes();
                    varlen_offsets.push(varlen_data.len() as u16);

                    varlen_data.extend_from_slice(&(ts_bytes.len() as u32).to_le_bytes());
                    varlen_data.extend_from_slice(ts_bytes);
                    has_varlen = true;

                    fixed_data.extend_from_slice(&(varlen_offsets.len() as u32 - 1).to_le_bytes());
                }
            }
        }

        // 设置标志位
        if has_null {
            tuple.header.flags.set_has_null();
        }
        if has_varlen {
            tuple.header.flags.set_has_varlen();
            tuple.header.varlen_offset = (TupleHeaderV2::HEADER_SIZE
                + tuple.null_bitmap.len()
                + fixed_data.len()) as u16;
        }

        // 设置数据
        tuple.fixed_data = fixed_data;
        tuple.varlen_data = varlen_data;
        tuple.varlen_offsets = varlen_offsets;

        // 计算总大小
        tuple.header.total_size = tuple.total_size() as u16;

        // 更新统计信息
        let encoding_time = start_time.elapsed().as_micros() as u64;
        self.update_encoding_stats(encoding_time, tuple.total_size());

        Ok(tuple)
    }

    /// 解码元组为值向量
    ///
    /// 高效的解码实现，支持零拷贝优化
    pub fn decode_tuple(&mut self, tuple: &TupleV2) -> Result<Vec<Value>> {
        let start_time = std::time::Instant::now();

        let mut values = Vec::with_capacity(tuple.header.column_count as usize);
        let mut fixed_offset = 0;

        for (i, column) in self.schema.columns.iter().enumerate() {
            if tuple.is_null(i) {
                values.push(Value::Null);
                // 跳过定长数据中的占位符
                self.skip_fixed_data(&mut fixed_offset, &column.data_type);
                continue;
            }

            let value = match column.data_type {
                DataType::Integer => {
                    if fixed_offset + 8 > tuple.fixed_data.len() {
                        return Err(Error::storage("定长数据不足".to_string()));
                    }
                    let bytes = &tuple.fixed_data[fixed_offset..fixed_offset + 8];
                    let val = i64::from_le_bytes(bytes.try_into().unwrap());
                    fixed_offset += 8;
                    Value::Integer(val)
                }
                DataType::Real => {
                    if fixed_offset + 8 > tuple.fixed_data.len() {
                        return Err(Error::storage("定长数据不足".to_string()));
                    }
                    let bytes = &tuple.fixed_data[fixed_offset..fixed_offset + 8];
                    let val = f64::from_le_bytes(bytes.try_into().unwrap());
                    fixed_offset += 8;
                    Value::Real(val)
                }
                DataType::Boolean => {
                    if fixed_offset + 1 > tuple.fixed_data.len() {
                        return Err(Error::storage("定长数据不足".to_string()));
                    }
                    let val = tuple.fixed_data[fixed_offset] != 0;
                    fixed_offset += 1;
                    Value::Boolean(val)
                }
                DataType::Text(_) => {
                    // 从定长区域读取变长数据索引
                    if fixed_offset + 4 > tuple.fixed_data.len() {
                        return Err(Error::storage("定长数据不足".to_string()));
                    }
                    let bytes = &tuple.fixed_data[fixed_offset..fixed_offset + 4];
                    let varlen_index = u32::from_le_bytes(bytes.try_into().unwrap()) as usize;
                    fixed_offset += 4;

                    // 从变长数据区域读取实际数据
                    self.read_varlen_text(&tuple.varlen_data, &tuple.varlen_offsets, varlen_index)?
                }
                DataType::Decimal(_) => {
                    // 从变长数据读取Decimal
                    if fixed_offset + 4 > tuple.fixed_data.len() {
                        return Err(Error::storage("定长数据不足".to_string()));
                    }
                    let bytes = &tuple.fixed_data[fixed_offset..fixed_offset + 4];
                    let varlen_index = u32::from_le_bytes(bytes.try_into().unwrap()) as usize;
                    fixed_offset += 4;

                    self.read_varlen_decimal(&tuple.varlen_data, &tuple.varlen_offsets, varlen_index)?
                }
                DataType::Timestamp => {
                    // 从变长数据读取Timestamp
                    if fixed_offset + 4 > tuple.fixed_data.len() {
                        return Err(Error::storage("定长数据不足".to_string()));
                    }
                    let bytes = &tuple.fixed_data[fixed_offset..fixed_offset + 4];
                    let varlen_index = u32::from_le_bytes(bytes.try_into().unwrap()) as usize;
                    fixed_offset += 4;

                    self.read_varlen_timestamp(&tuple.varlen_data, &tuple.varlen_offsets, varlen_index)?
                }
                DataType::BigInt => {
                    if fixed_offset + 8 > tuple.fixed_data.len() {
                        return Err(Error::storage("定长数据不足".to_string()));
                    }
                    let bytes = &tuple.fixed_data[fixed_offset..fixed_offset + 8];
                    let val = i64::from_le_bytes(bytes.try_into().unwrap());
                    fixed_offset += 8;
                    Value::Integer(val)
                }
                DataType::Double => {
                    if fixed_offset + 8 > tuple.fixed_data.len() {
                        return Err(Error::storage("定长数据不足".to_string()));
                    }
                    let bytes = &tuple.fixed_data[fixed_offset..fixed_offset + 8];
                    let val = f64::from_le_bytes(bytes.try_into().unwrap());
                    fixed_offset += 8;
                    Value::Real(val)
                }
                DataType::Date => {
                    // Date作为变长数据存储
                    if fixed_offset + 4 > tuple.fixed_data.len() {
                        return Err(Error::storage("定长数据不足".to_string()));
                    }
                    let bytes = &tuple.fixed_data[fixed_offset..fixed_offset + 4];
                    let varlen_index = u32::from_le_bytes(bytes.try_into().unwrap()) as usize;
                    fixed_offset += 4;

                    self.read_varlen_text(&tuple.varlen_data, &tuple.varlen_offsets, varlen_index)?
                }
            };

            values.push(value);
        }

        // 更新统计信息
        let decoding_time = start_time.elapsed().as_micros() as u64;
        self.update_decoding_stats(decoding_time);

        Ok(values)
    }

    /// 获取编码统计信息
    pub fn get_stats(&self) -> &TupleEncodingStats {
        &self.stats
    }

    /// 重置统计信息
    pub fn reset_stats(&mut self) {
        self.stats = TupleEncodingStats::default();
    }

    /// 为定长类型添加占位符
    fn add_fixed_placeholder(&self, fixed_data: &mut Vec<u8>, data_type: &DataType) {
        match data_type {
            DataType::Integer => fixed_data.extend_from_slice(&[0u8; 8]),
            DataType::BigInt => fixed_data.extend_from_slice(&[0u8; 8]),
            DataType::Real => fixed_data.extend_from_slice(&[0u8; 8]),
            DataType::Double => fixed_data.extend_from_slice(&[0u8; 8]),
            DataType::Boolean => fixed_data.push(0),
            DataType::Text(_) | DataType::Decimal(_) | DataType::Timestamp | DataType::Date => {
                // 变长类型在定长区域存储索引
                fixed_data.extend_from_slice(&[0u8; 4]);
            }
        }
    }

    /// 跳过定长数据中的占位符
    fn skip_fixed_data(&self, offset: &mut usize, data_type: &DataType) {
        match data_type {
            DataType::Integer => *offset += 8,
            DataType::BigInt => *offset += 8,
            DataType::Real => *offset += 8,
            DataType::Double => *offset += 8,
            DataType::Boolean => *offset += 1,
            DataType::Text(_) | DataType::Decimal(_) | DataType::Timestamp | DataType::Date => *offset += 4,
        }
    }

    /// 从变长数据读取文本
    fn read_varlen_text(
        &self,
        varlen_data: &[u8],
        varlen_offsets: &[u16],
        index: usize,
    ) -> Result<Value> {
        if index >= varlen_offsets.len() {
            return Err(Error::storage("变长数据索引越界".to_string()));
        }

        let offset = varlen_offsets[index] as usize;
        if offset + 4 > varlen_data.len() {
            return Err(Error::storage("变长数据偏移越界".to_string()));
        }

        // 读取长度
        let len_bytes = &varlen_data[offset..offset + 4];
        let len = u32::from_le_bytes(len_bytes.try_into().unwrap()) as usize;

        // 读取数据
        let data_start = offset + 4;
        if data_start + len > varlen_data.len() {
            return Err(Error::storage("变长数据长度越界".to_string()));
        }

        let text_bytes = &varlen_data[data_start..data_start + len];
        let text = String::from_utf8(text_bytes.to_vec())
            .map_err(|e| Error::storage(format!("UTF-8解码错误: {}", e)))?;

        Ok(Value::Text(text))
    }

    /// 从变长数据读取Decimal
    fn read_varlen_decimal(
        &self,
        varlen_data: &[u8],
        varlen_offsets: &[u16],
        index: usize,
    ) -> Result<Value> {
        if let Value::Text(text) = self.read_varlen_text(varlen_data, varlen_offsets, index)? {
            let decimal = text.parse::<rust_decimal::Decimal>()
                .map_err(|e| Error::storage(format!("Decimal解析错误: {}", e)))?;
            Ok(Value::Decimal(decimal))
        } else {
            Err(Error::storage("Decimal数据格式错误".to_string()))
        }
    }

    /// 从变长数据读取Timestamp
    fn read_varlen_timestamp(
        &self,
        varlen_data: &[u8],
        varlen_offsets: &[u16],
        index: usize,
    ) -> Result<Value> {
        if let Value::Text(text) = self.read_varlen_text(varlen_data, varlen_offsets, index)? {
            let timestamp = chrono::DateTime::parse_from_rfc3339(&text)
                .map_err(|e| Error::storage(format!("Timestamp解析错误: {}", e)))?;
            Ok(Value::Timestamp(timestamp.with_timezone(&chrono::Utc)))
        } else {
            Err(Error::storage("Timestamp数据格式错误".to_string()))
        }
    }

    /// 更新编码统计信息
    fn update_encoding_stats(&mut self, encoding_time_us: u64, tuple_size: usize) {
        self.stats.encoded_tuples += 1;
        self.stats.total_encoding_time_us += encoding_time_us;
        self.stats.uncompressed_size += tuple_size as u64;

        // 计算平均编码时间
        if self.stats.encoded_tuples > 0 {
            self.stats.avg_encoding_time_us =
                self.stats.total_encoding_time_us / self.stats.encoded_tuples;
        }

        self.stats.last_updated = SystemTime::now();
    }

    /// 更新解码统计信息
    fn update_decoding_stats(&mut self, decoding_time_us: u64) {
        self.stats.decoded_tuples += 1;
        self.stats.total_decoding_time_us += decoding_time_us;

        // 计算平均解码时间
        if self.stats.decoded_tuples > 0 {
            self.stats.avg_decoding_time_us =
                self.stats.total_decoding_time_us / self.stats.decoded_tuples;
        }

        self.stats.last_updated = SystemTime::now();
    }
}

/// 元组压缩器 - 支持高效的元组压缩
#[derive(Debug)]
pub struct TupleCompressor {
    /// 压缩算法类型
    compression_type: CompressionType,
    /// 压缩统计信息
    stats: CompressionStats,
}

/// 压缩算法类型
#[derive(Debug, Clone, Copy)]
pub enum CompressionType {
    /// 无压缩
    None,
    /// LZ4压缩（快速）
    Lz4,
    /// Zstd压缩（高压缩比）
    Zstd,
    /// 字典压缩（适合重复数据）
    Dictionary,
}

/// 压缩统计信息
#[derive(Debug, Clone, Default)]
pub struct CompressionStats {
    /// 压缩的元组数量
    pub compressed_tuples: u64,
    /// 解压的元组数量
    pub decompressed_tuples: u64,
    /// 压缩前总大小
    pub original_size: u64,
    /// 压缩后总大小
    pub compressed_size: u64,
    /// 总压缩时间（微秒）
    pub total_compression_time_us: u64,
    /// 总解压时间（微秒）
    pub total_decompression_time_us: u64,
    /// 平均压缩比
    pub avg_compression_ratio: f64,
    /// 平均压缩时间（微秒）
    pub avg_compression_time_us: u64,
    /// 平均解压时间（微秒）
    pub avg_decompression_time_us: u64,
}

impl TupleCompressor {
    /// 创建新的元组压缩器
    pub fn new(compression_type: CompressionType) -> Self {
        Self {
            compression_type,
            stats: CompressionStats::default(),
        }
    }

    /// 压缩元组数据
    pub fn compress(&mut self, data: &[u8]) -> Result<Vec<u8>> {
        let start_time = std::time::Instant::now();
        let original_size = data.len();

        let compressed_data = match self.compression_type {
            CompressionType::None => data.to_vec(),
            CompressionType::Lz4 => {
                // 简化实现：使用基础压缩算法
                self.simple_compress(data)
            }
            CompressionType::Zstd => {
                // 简化实现：使用基础压缩算法
                self.simple_compress(data)
            }
            CompressionType::Dictionary => {
                // 简化实现：使用字典压缩
                self.dictionary_compress(data)
            }
        };

        let compression_time = start_time.elapsed().as_micros() as u64;
        self.update_compression_stats(original_size, compressed_data.len(), compression_time);

        Ok(compressed_data)
    }

    /// 解压元组数据
    pub fn decompress(&mut self, data: &[u8]) -> Result<Vec<u8>> {
        let start_time = std::time::Instant::now();

        let decompressed_data = match self.compression_type {
            CompressionType::None => data.to_vec(),
            CompressionType::Lz4 => {
                // 简化实现：使用基础解压算法
                self.simple_decompress(data)?
            }
            CompressionType::Zstd => {
                // 简化实现：使用基础解压算法
                self.simple_decompress(data)?
            }
            CompressionType::Dictionary => {
                // 简化实现：使用字典解压
                self.dictionary_decompress(data)?
            }
        };

        let decompression_time = start_time.elapsed().as_micros() as u64;
        self.update_decompression_stats(decompression_time);

        Ok(decompressed_data)
    }

    /// 获取压缩统计信息
    pub fn get_stats(&self) -> &CompressionStats {
        &self.stats
    }

    /// 简单压缩算法（RLE压缩）
    fn simple_compress(&self, data: &[u8]) -> Vec<u8> {
        let mut compressed = Vec::new();

        if data.is_empty() {
            return compressed;
        }

        let mut i = 0;
        while i < data.len() {
            let current_byte = data[i];
            let mut count = 1;

            // 计算连续相同字节的数量
            while i + count < data.len() && data[i + count] == current_byte && count < 255 {
                count += 1;
            }

            // 存储计数和字节值
            compressed.push(count as u8);
            compressed.push(current_byte);

            i += count;
        }

        compressed
    }

    /// 简单解压算法
    fn simple_decompress(&self, data: &[u8]) -> Result<Vec<u8>> {
        let mut decompressed = Vec::new();

        let mut i = 0;
        while i + 1 < data.len() {
            let count = data[i] as usize;
            let byte_value = data[i + 1];

            for _ in 0..count {
                decompressed.push(byte_value);
            }

            i += 2;
        }

        Ok(decompressed)
    }

    /// 字典压缩（简化实现）
    fn dictionary_compress(&self, data: &[u8]) -> Vec<u8> {
        // 简化实现：直接返回原数据
        data.to_vec()
    }

    /// 字典解压（简化实现）
    fn dictionary_decompress(&self, data: &[u8]) -> Result<Vec<u8>> {
        // 简化实现：直接返回原数据
        Ok(data.to_vec())
    }

    /// 更新压缩统计信息
    fn update_compression_stats(&mut self, original_size: usize, compressed_size: usize, time_us: u64) {
        self.stats.compressed_tuples += 1;
        self.stats.original_size += original_size as u64;
        self.stats.compressed_size += compressed_size as u64;
        self.stats.total_compression_time_us += time_us;

        // 计算平均值
        if self.stats.compressed_tuples > 0 {
            self.stats.avg_compression_ratio =
                self.stats.compressed_size as f64 / self.stats.original_size as f64;
            self.stats.avg_compression_time_us =
                self.stats.total_compression_time_us / self.stats.compressed_tuples;
        }
    }

    /// 更新解压统计信息
    fn update_decompression_stats(&mut self, time_us: u64) {
        self.stats.decompressed_tuples += 1;
        self.stats.total_decompression_time_us += time_us;

        if self.stats.decompressed_tuples > 0 {
            self.stats.avg_decompression_time_us =
                self.stats.total_decompression_time_us / self.stats.decompressed_tuples;
        }
    }
}
